package com.yh.csx.bsf.ods;

import com.yh.csx.bsf.core.db.DbConn;
import com.yh.csx.bsf.core.util.ConvertUtils;
import com.yh.csx.bsf.core.util.FileUtils;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.core.util.NetworkUtils;
import lombok.var;

import java.io.File;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class CanalServerProvider extends AbstractCanalServerProvider {

    @Override
    public void start(){
        stop();
        //解压文件夹
        unzip(false);
        LogUtils.info(this.getClass(),"ods","canal server 已解压,不覆盖!");
        inputProperty(false);
        LogUtils.info(this.getClass(),"ods","canal server 初始化配置!");
        super.start();
    }


    private void inputProperty(boolean overridOldConf){
        //配置canal.properties
        SafeProperties sp = new SafeProperties();
        var file = new File("canal/conf/canal.properties");
        sp.load(file);
        sp.setProperty("canal.id", NetworkUtils.getIpAddress().replace(".",""));
        sp.setProperty("canal.instance.filter.query.dcl","true");
        sp.setProperty("canal.instance.filter.query.dml","true");
        sp.setProperty("canal.instance.filter.query.ddl","true");
        sp.setProperty("canal.instance.tsdb.enable","false");
   

        Set<String> destinations = new HashSet<String>();
        
        Map<String,String> servers=new HashMap<String,String>();
		for (var db : CommonUtils.getSubscribeDbs()) {
			Map<String, String> mapInfo=stragtegyProvider.getDbInfo(db);
			servers.put(mapInfo.get("address"),db);
			
		}
		for (String db : servers.keySet()) {		
			destinations.add(db);
            //复制文件夹
        	File oldConf = new File("canal/conf/"+db+"/instance.properties");
        	if(!oldConf.exists()) {
	            FileUtils.copyDir(new File("canal/example").getAbsolutePath(),new File("canal/conf/"+db).getAbsolutePath());
	            LogUtils.info(this.getClass(),"ods","canal server 开始初始化"+db+"配置!");
	            try(DbConn dbConn = stragtegyProvider.getDb(servers.get(db))) {
	                if(dbConn==null)
	                    throw new RuntimeException("未配置数据库数据源:"+db);
	                var dbinfo = stragtegyProvider.getDbInfo(servers.get(db));
	                var rs = dbConn.executeList("show master status", null);
	                if (rs.size() == 0) {
	                    throw new RuntimeException("数据库不支持主从同步");
	                }
	
	                var journalName = rs.get(0).getOrDefault("File", "");
	                var position = rs.get(0).getOrDefault("Position", "");
	                SafeProperties sp2 = new SafeProperties();
	                var file2 = new File("canal/conf/"+db+"/instance.properties");
	                if(!file2.exists())
	                {
	                    LogUtils.warn(this.getClass(),"ods","配置文件找不到:"+file2.getAbsolutePath());
	                   
	                }
	                sp2.load(file2);
	                sp2.setProperty("canal.instance.filter.regex", ".*\\..*");
	                sp2.setProperty("canal.instance.tsdb.enable", "false");
	                sp2.setProperty("canal.instance.master.address", dbinfo.get("address").replace("\\", ""));
	                sp2.setProperty("canal.instance.master.journal.name", ConvertUtils.convert(journalName, String.class));
	                sp2.setProperty("canal.instance.master.position", ConvertUtils.convert(position, String.class));
	                sp2.setProperty("canal.instance.dbUsername", dbinfo.get("username"));
	                sp2.setProperty("canal.instance.dbPassword", dbinfo.get("password"));
	                sp2.store(file2);
	                
	               
	                LogUtils.info(this.getClass(),"ods","canal server 结束初始化"+db+"配置!");
	            }
        	}
		}
        sp.setProperty("canal.destinations",String.join(",",destinations));
        sp.store(file);
    }
}
